package com.mate.cloud;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

import java.nio.charset.StandardCharsets;
import java.util.List;

/**
 * @description:    RocketMQ SDK同步消费消息测试
 * @author:         MI
 * @email:          448341911@qq.com
 * @createTime:     2025/7/2 18:52
 * @updateUser:     MI
 * @updateTime:     2025/7/2 18:52
 * @updateRemark:   修改内容
 * @version:        1.0
 */
public class RocketMqSyncConsumer {
    public static void main(String[] args) throws MQClientException, InterruptedException {
        // 实例化消费者组名称
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("order_sync_group_consumer");
        // 指定name server地址
        consumer.setNamesrvAddr("192.168.110.17:9876");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        // 订阅至少一个主题以供消费
        consumer.subscribe("order_sync_topic", "*");
        //  注册回调，处理从服务端获取的消息
        consumer.registerMessageListener((List<MessageExt> msgs, ConsumeConcurrentlyContext context) -> {
            try {
                for (MessageExt msg : msgs) {
                    String body = new String(msg.getBody(), StandardCharsets.UTF_8);
                    System.out.printf("线程 [%s] 收到消息：msgId=%s, body=%s%n",
                            Thread.currentThread().getName(),
                            msg.getMsgId(),
                            body);
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            } catch (Exception e) {
                e.printStackTrace();
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        });

        // 启动消费者实例
        consumer.start();
        System.out.println("消费者已启动");
    }
}
